import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.*;

public class KafkaProducer2 {
    public static void main(String[] args) {
        // 1. 创建 Kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 配置 Kafka 参数
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.23.128:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 初始化课程、班级和学生数据
        String[] courseArray = {"数学", "英语", "物理", "化学", "生物"};
        List<String> classIds = Arrays.asList("220711", "220712", "220713", "220714", "220715"); // 固定班级列表
        Map<String, Map<String, String>> classStudentData = new HashMap<>();

        // 初始化班级和学生数据
        for (String classId : classIds) {
            Map<String, String> students = new HashMap<>();
            // 根据班级生成不同的学号前缀
            String studentIdPrefix;
            switch (classId) {
                case "220711":
                    studentIdPrefix = "2022340701";
                    break;
                case "220712":
                    studentIdPrefix = "2022340702";
                    break;
                case "220713":
                    studentIdPrefix = "2022340703";
                    break;
                case "220714":
                    studentIdPrefix = "2022340704";
                    break;
                case "220715":
                    studentIdPrefix = "2022340705";
                    break;
                default:
                    studentIdPrefix = "2022340700"; // 默认前缀，可根据需要修改
                    break;
            }
            for (int j = 1; j < 30; j++) {
                String studentName = "学生" + j;
                String studentId = studentIdPrefix + String.format("%02d", new Random().nextInt(30) + 1);
                students.put(studentName, studentId);
            }
            classStudentData.put(classId, students);
        }

        // 3. 创建 Kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        while (true) {
            System.out.println("开始发送数据==========================");
            try {
                // 4. 循环推送 5 条数据
                for (int i = 0; i < 5; i++) {
                    // 随机选择班级
                    String classId = classIds.get(new Random().nextInt(classIds.size()));
                    // 随机选择学生
                    Map<String, String> students = classStudentData.get(classId);
                    List<String> studentNames = new ArrayList<>(students.keySet());
                    String studentName = studentNames.get(new Random().nextInt(studentNames.size()));
                    String studentId = students.get(studentName);
                    // 随机选择课程和状态
                    String courseName = courseArray[new Random().nextInt(courseArray.length)];
                    String attendanceStatus = new Random().nextBoolean() ? "A" : "L";
                    // 拼接记录
                    String record = String.join("\t", classId, studentName, courseName, studentId, attendanceStatus);
                    // 发送到 Kafka
                    kafkaProducer.send(new ProducerRecord<>("attendance123", record));
                    System.out.println("发送数据: " + record);
                }
                // 间隔 5 秒
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
